package com.bee.shepherd.config;

import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;

@Configuration
@PropertySource(value = "classpath:kafka.properties")
public class KafkaConfig {

    @Value("${kafka.consumer.group.id}")
    private String groupId;
    @Value("${kafka.consumer.bootstrap.servers}")
    private String bootStrapServers;
    @Value("${kafka.consumer.enable.auto.commit}")
    private boolean autoCommit;
    @Value("${kafka.consumer.auto.commit.interval.ms}")
    private int autoCommitIntervalMs;
    @Value("${kafka.consumer.key-deserializer}")
    private String keyDeSerializer;
    @Value("${kafka.consumer.value-deserializer}")
    private String valueDeSerializer;
    @Value("${kafka.consumer.auto.offset.reset}")
    private String offsetReset;
    @Value("${kafka.consumer.topic}")
    private String topic;
    @Value("${kafka.consumer.sasl.mechanism}")
    private String saslMechanism;
    @Value("${kafka.consumer.security.protocol}")
    private String secProtocol;
    @Value("${kafka.consumer.sasl.jaas.config}")
    private String jaasConfig;

    @Bean
    public JAASKafkaConsumerConnector kafkaCanalConnector(){
        return new JAASKafkaConsumerConnector(bootStrapServers,topic,null,groupId,50,
                true,secProtocol,saslMechanism,jaasConfig);
    }

    public class JAASKafkaConsumerConnector extends KafkaCanalConnector {
        JAASKafkaConsumerConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
                                   boolean flatMessage,String secProtocol,String mechanism,String jaasConfig){
            super(servers,topic,partition,groupId,batchSize,flatMessage);
            properties.put("security.protocol",secProtocol);
            properties.put("sasl.mechanism",mechanism);
            properties.put("sasl.jaas.config",jaasConfig);
        }
    }

}
